using namespace std;

#include "heap.h"
#include "hash.h"
#include "connection.h"
#include "exchange.h"
#include "worker.h"
#include "test.h"
#include "port.h"
#include "scheduler.h"

#include <stdlib.h>
#include <time.h>
#include <iostream>
#include <fstream>
#include <pthread.h>
#include <unistd.h>
#include <sys/select.h>

/****************************************/
// Global variables, shared among threads

pthread_mutex_t** socket_semaphore;
pthread_mutex_t** cq_semaphore;
pthread_mutex_t** portmutex;
HashTable<int> connection_map(MAXCONN);
HashTable<int> sock2port(MAXCONN);
HashTable<Connection> connections(MAXCONN);
Heap<Exchange>* ce_master;
Heap<Exchange>** connexchanges;
Heap<Exchange>** dataexchanges;
Heap<Port>* worker_ports;
fd_set* sockets;
int* maxfd, maxID;
int* condone, *conorig, total;
double* diff, *maxdiff;
double maxtime=0;
HashTable<Port> ports(MAXP);
int* stalled;
HashTable<int> prefixes(MAXPREFIX);
/****************************************/


// Utility functions
int getpeerID(string line)
{
  char* token, contents[MAXL];

  strcpy(contents, line.c_str());
  token = strtok(contents, " ");
  int pass = 0;
  while (token != NULL)
    {
      switch(pass)
	{
	case 7: 
	  return atoi(token);
	}
      pass++;
      token = strtok(NULL, " ");
    }
}

void usage(char* prog)
{
  cout << "Usage "<<prog<<" log file name and number of machines"<<endl;
}

void* runScheduler(void* ptr)
{
  Scheduler s;
  s.Run();
}

void* runWorker(void* ptr)
{
  int ID = *((int*) ptr);
  Worker w(ID);
  w.Run();
}

int Open(Port p)
{
  int sockfd, on;
  struct addrinfo my_addr, *res;
  char port[MAXL];

  if ((sockfd = socket(AF_INET, SOCK_STREAM, 0)) == -1)
    {
      perror("socket");
      return NONE;
    }

  if (setsockopt (sockfd, SOL_SOCKET, SO_REUSEADDR,(void *) &on, sizeof (on)) == -1)
    {
      perror("setsockopt");
      return NONE;
    }

  memset(&my_addr, 0, sizeof(my_addr));
  my_addr.ai_family = AF_UNSPEC;
  my_addr.ai_socktype = SOCK_STREAM;
  my_addr.ai_flags = AI_PASSIVE;

  sprintf(port, "%d", p.GetPort());
  getaddrinfo(NULL, port, &my_addr, &res);
  if (bind(sockfd, res->ai_addr, res->ai_addrlen) == -1)
    {
      char errmsg[MAXL];
      sprintf(errmsg, "Bind port %d\n", p.GetPort());
      perror(errmsg);
      return NONE;
    }
  if (listen(sockfd, MAX_CLIENTS) == -1)
    {
      perror("listen");
      return NONE;
    }
  sock2port.insert(sockfd, new int(p.GetPort()));
  return sockfd;
}

void* listen(void* ptr)
{
  int starttime = time(0), wID = 0;
  int tc = *((int*) ptr);
  int connsperworker = tc/MAXWORKERS;

  cout << "Listen thread started at "<<starttime<<endl;
  while(TRUE)
    {      
      int now = time(0) - starttime;
      int totleft = total, count = 0;
      double tdiff = 0, tmaxdiff = 0;
      if (now > maxtime + WAIT)
	break;
      for(int i=0; i<MAXWORKERS; i++)
	{
	  totleft=totleft - condone[i];
	  count = count + condone[i];
	  tdiff = tdiff + diff[i];
	  if (maxdiff[i] > tmaxdiff)
	    tmaxdiff = maxdiff[i];
	}	  
      if (count == 0)
	count = 1;

      cout << "LT: "<<now << " Listen thread, still left " << totleft << " will exit at "<<(maxtime+WAIT)<<" maxdiff "<<tmaxdiff<<" avg diff "<<tdiff/count<<endl;
      while(worker_ports->HasMore())
	{
	  Port p = worker_ports->peak();

	  // Open ports a little sooner than needed
	  if (p.GetTime() - LEEWAY > now)
	    break;
	  worker_ports->pop();
	  int sockfd = Open(p);
	  if (sockfd != NONE)	    
	    {
	      cout << "LT: "<< time(0)-starttime << " absolute time "<< time(0)<<" Opened port " << p << endl;	  
	      int rest = p.GetConnectionCount();
	      while (rest > 0)
		{
		  int qID = wID % MAXWORKERS;
		  wID++;				
		  pthread_mutex_lock(socket_semaphore[qID]);
		  FD_SET(sockfd,&sockets[qID]);	  
		  if (sockfd > maxfd[qID])
		    maxfd[qID] = sockfd;
		  pthread_mutex_unlock(socket_semaphore[qID]);
		  rest = rest - (tc/MAXWORKERS+1);
		}
	    }
	  else
	    cout << "Couldn't open "<< p << endl;
	}
      cout << now << " Listen thread exited" << endl;
      usleep(1000);
    }
}

int loadFile(ifstream& infile)
{
  int wID = 0, tc;
  while(TRUE)
    {
      string line;

      getline(infile, line);
      if (infile.eof())
	break;

      Exchange* e = new Exchange(line);
      if (e->time > maxtime)
	maxtime = e->time;
      if (e->type == 'A' || e->type == 'C')
	{
	  Connection* c = new Connection(line);
	  connections.insert(c->ID, c);
	  cout<<"INserted connection with ID "<<c->ID<<" lookup returns "<<connections.lookup(c->ID)<<endl;
	  prefixes.insert(c->myIP & 0x00ffffff, new int(c->myIP & 0x00ffffff));
	  e->c = c;
	  if (c->ID > maxID)
	    maxID = c->ID;
	  c->AddExchange(*e);
	  if (e->type == 'A')
	    {
	      Port* p = ports.lookup(c->port);
	      if (!p)
		{
		  p = new Port(c->port, c->list[0].time);
		  ports.insert(c->port, p);
		}
	      else
		p->UpdateStarttime(c->list[0].time);
	      p->AddConnection();
	    }
	  wID++;
	}
      else
	{ 
	  Connection* c = connections.lookup(e->myID);
	  c->AddExchange(*e);
	  if (e->type == 'P')
	    {
	      c->peerID = getpeerID(line);
	    }
	}
    }

  /* Count all prefixes */
  int *pr, pcount = 0;
  for(prefixes.begin(); pr=prefixes.get(); )
    pcount++;

  cout <<"Prefixes "<<pcount<<endl;

  /* Go through all connections and insert FIN packets. 
     Also get rid of SYN-FIN connections */

  Connection* c;
  for(connections.begin(); c=connections.get(); )
    {
      if (c->spot == 1)
	connections.remove(c->ID);
      else
	{
	  Exchange first = c->list[0], last = c->list[c->listsize -1];
	  Exchange* ne = new Exchange(first.time, 'F', 0, c->ID, c);
	 if (c->peerID <= 0 && first.type == 'C')
	    cout << "Connection "<<c->ID<<" doesn't have a peer ID listsize is "<<c->spot<<endl;

	  //cout << "Added FIN to connection "<<c->ID<<endl;
	  c->AddExchange(*ne);
	  if (first.type == 'C')
	    {
	      ce_master->insert(first);
	    }
	}
    }
  // Go through all ports now and insert into a heap.
  // Couldn't do this before because we had no conn counts.
  Port* p;
  wID = 0;
  tc = 0;
  for(ports.begin(); (p = ports.get()) != NULL;)
    {
      worker_ports->insert(*p); 
      tc = tc + p->GetConnectionCount();
    }
  worker_ports->print();
  return tc;
}


int main(int argc, char** argv)
{
  pthread_t thread[MAXWORKERS], listenthread, scheduler;
  if (argc < 3)
    {
      usage(argv[0]);
      exit(0);
    }
  ifstream infile;
  infile.open(argv[1]);
  if (!infile)
    {
      cerr << "No such file "<< argv[1];
      exit(0);
    }
  int slaves = atoi(argv[2]) - 1, i;

  // Allocate heaps
  ce_master = new Heap<Exchange> (MAXCEX+1);
  connexchanges = new Heap<Exchange>* [MAXWORKERS];
  dataexchanges = new Heap<Exchange>* [MAXWORKERS];
  worker_ports = new Heap<Port> [MAXP];

  // Sockets and semaphores
  sockets = new fd_set[MAXWORKERS];
  maxfd = new int[MAXWORKERS];
  stalled = new int[MAXWORKERS];
  socket_semaphore = new pthread_mutex_t*[MAXWORKERS];
  cq_semaphore = new pthread_mutex_t*[MAXWORKERS];
  portmutex = new pthread_mutex_t*[MAXPORTS];

  // Counter to keep track of what is left to be done
  condone = new int[MAXWORKERS];
  conorig = new int[MAXWORKERS];
  diff = new double[MAXWORKERS];
  maxdiff = new double[MAXWORKERS];

  for(i=0; i<MAXPORTS; i++)
    {
      pthread_mutex_t *p = new pthread_mutex_t;
      pthread_mutex_init(p,NULL);
      portmutex[i] = p;
    }

  for(i=0; i<MAXWORKERS; i++)
    {
      connexchanges[i] = new Heap<Exchange> (MAXCEX/MAXWORKERS+1);
      dataexchanges[i] = new Heap<Exchange> (MAXDEX/MAXWORKERS+1);
      pthread_mutex_t *s = new pthread_mutex_t;
      pthread_mutex_init(s,NULL);
      socket_semaphore[i] = s;
      pthread_mutex_t *cq = new pthread_mutex_t;
      pthread_mutex_init(cq,NULL);
      cq_semaphore[i] = cq;
      FD_ZERO(&sockets[i]);
      maxfd[i] = -1;
    }  
  int tc = loadFile(infile);

  /* Perform randez-vous among machines */
  char hostname[MAXL];
  gethostname(hostname, MAXL);
  cout << "Hostname is "<<hostname<<endl;
  if (strncmp(hostname,"node1.",6) == 0)
    {
      int s = slaves;
      i = 0;
      cout<<"This is node1"<<endl;
      Port sync(SYNCPORT,0);
      int sockfd = Open(sync);
      int* newfd = new int(slaves);
      while(s > 0)
	{
	  struct sockaddr their_addr;
	  socklen_t addr_size = sizeof(their_addr);
	  newfd[i++] = accept(sockfd, (struct sockaddr*) &their_addr, &addr_size);
	  cout << "Waiting for "<< s <<" sockfd "<<sockfd<<" new "<<newfd[i]<<endl;
	  s--;
	}
      char buf[] = "OK";
      int n;
      for (i=0;i<slaves;i++)
	{
	  n=send(newfd[i], buf, 3, 0);
	  close(newfd[i]);
	  cout << "Sent "<<n<<" bytes to sock "<<newfd[i]<<endl;
	}
    }
  else
    {
      struct hostent* he = gethostbyname("node1");
      struct in_addr** master_addr_list = (struct in_addr **)he->h_addr_list;
      int masterIP = master_addr_list[0]->s_addr;
      struct hostent* me = gethostbyname(hostname);
      struct in_addr** my_addr_list = (struct in_addr **)me->h_addr_list;
      int myIP = my_addr_list[0]->s_addr;
      Connection c(0, myIP, masterIP, SYNCPORT);
      int sockfd = 0;
      while (!sockfd)
	{
	  sockfd = c.Connect();
	  cout << "socket "<<sockfd<<endl;
	  sleep(1);
	}
      char buf[MAXL];
      int n = recv(sockfd, buf, MAXL, 0);
      cout << "Received "<<n<<" bytes, message "<<buf<<endl;
      close(sockfd);
    }
  pthread_create(&listenthread, NULL, &listen, &tc);
  sleep(3);
  for(int i=0; i<MAXWORKERS; i++)
    {
      int *m = new int;
      *m = i;
      pthread_create(&thread[i], NULL, &runWorker, m);
    }
  pthread_create(&scheduler, NULL, &runScheduler, NULL);
  for(int i=0; i<MAXWORKERS; i++)
    pthread_join(thread[i], NULL);

}
